The _PIPELINE_DONE sentinel is pushed to the queue, but the consumption logic in the finally block or the iterator loop does not explicitly handle it as a termination signal for individual threads, relying instead on the queue being drained. Ensure this is the intended design to avoid potential race conditions.

Should all the turns be ordered and consecutive in the dataset? For instance if i have turn 1 of all conversations followed by turn 2 of all conversations - is that a valid format. Would it not be easier to have a conversation as a top level object which contains a list of messages and associated metadata inside?
The challenge here would be reading in all the conversations to collect all the turns - unless we enforce some constraints.

Yes, all turns are now ordered and consecutive.

Ideally, these should be relative and follow the readme example. So if you follow the readme example, this config should work out of the box.

🟠 HIGH · bug · [Claude]

from_endpoint_response violates the QueryResult.metadata type contract by passing metadata=None when no tool_calls/finish_reason/reasoning_content are present. QueryResult.metadata is typed dict[str, Any] with default_factory=dict. The constructed result is then passed to worker._handle_non_streaming_body which calls result.with_metadata(req.query_metadata). with_metadata short-circuits on empty additional_metadata (current callers pass {}), but as soon as any caller routes a non-empty Query.metadata (e.g., the conversation_id round-trip explicitly mentioned in Query.metadata's docstring), dict(self.metadata) will raise TypeError: 'NoneType' object is not iterable. Likewise, any code that does result.metadata.get(...) on this result will raise AttributeError. Fix:

return QueryResult(
    id=result_id or response.id,
    response_output=TextModelOutput(output=choice.message.content or ""),
    metadata=metadata,  # always pass dict; let omit_defaults handle empty
)

Fixed. metadata initialized to empty dict

🟠 HIGH · error-handling · [Claude]

await asyncio.gather(*tasks, return_exceptions=True) silently swallows every exception raised by a _conv_pipeline coroutine. The result list is discarded — there is no for r in results: if isinstance(r, Exception): .... Programmer bugs (KeyError on missing metadata key, AttributeError on result.metadata, dataset metadata-format drift, etc.) become invisible: the strategy returns phase_issuer.issued_count as if everything succeeded, and the only sign of failure is missing samples. AGENTS.md explicitly flags swallowed exceptions as forbidden. Fix: collect the result list and re-raise the first non-CancelledError exception, or at least log each one before discarding:

results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, BaseException) and not isinstance(r, asyncio.CancelledError):
        logger.error("conv_pipeline crashed: %r", r, exc_info=r)

Fixed. Errors are collected and logged.

🟡 MEDIUM · bug · [Claude]

When a turn times out waiting for the previous turn's response, the pipeline does state.failed_turns += 1 but never increments state.completed_turns, then breaks out of the loop. As a result, ConversationState.is_complete() (completed_turns >= expected_client_turns) returns False forever, the conversation never logs completion, and downstream _log_if_complete reporting silently drops it. Worse, if the late response arrives later, mark_turn_complete runs on the abandoned state and bumps completed_turns once, leaving completed_turns=1, failed_turns=1 for what was a multi-turn conversation, with the remaining turns never issued — but no error is raised. Fix: bump both counters on timeout (or call self._conv_manager.mark_turn_failed(...)), so the failure is properly accounted, and consider also draining/discarding the now-orphan _inflight entries for this conversation so a late response doesn't mutate state for a dead pipeline.

except TimeoutError:
    logger.warning(...)
    state.completed_turns += 1
    state.failed_turns += 1
    break

Fixed. Timeout calls mark_turn_failed() which bumps the counters.

🟡 MEDIUM · data-integrity · [Claude]

PromptData.text = data.get("prompt") and token_ids = data.get("input_tokens") or data.get("token_ids") — neither is set on multi-turn samples. MultiTurnDataset.load() writes the conversation into sample["messages"] (a list of message dicts) and does not populate prompt, input_tokens, or token_ids. The result is that every multi-turn ISSUED event records PromptData(text=None, token_ids=None), so the metrics aggregator computes ISL from zero or just the bare current-turn user text instead of the full prompt-with-history actually sent over the wire. ISL/throughput/TPOT-derived numbers reported for any multi-turn benchmark will therefore be wrong. Either serialize the merged messages into a synthetic prompt for ISL accounting, or extend PromptData to carry the OpenAI messages list and have the aggregator handle it. At minimum, document the limitation prominently and emit a one-shot warning when running multi-turn without a tokenizer.

Fixed. Contents are jointed into prompt_text

🟡 MEDIUM · bug · [Claude]

system_prompts_by_conv is keyed by str(conv_id) (line 212), but pre_built_messages_by_key and current_turn_messages_by_key are keyed by the raw conv_id returned from groupby (lines 266–267), and samples stores the raw value too. MultiTurnStrategy.execute() then looks up sys_prompts.get(conv_id) with the raw value (multi_turn_strategy.py line 116), so the system prompt is silently dropped whenever a JSONL has a non-string conversation_id (integer ids, mixed types after pandas type inference, etc.). Pick one canonicalization (preferably str(conv_id)) and apply it consistently across all dictionaries — including the samples entries and the _conv_states/_inflight keys in the strategy.

Fixed. All keys now use str(conv_id)

🔵 LOW · bug · [Claude]

When iterating prior_rows to build pre-built messages, a tool row whose tool_results field is an empty list (e.g. tool_results: []) is collected into msg, then _expand_tool_results(msg) returns [] (line 244 falls through), and the bare malformed msg (role='tool', no tool_call_id, no content, just the empty tool_results) is appended to the message history at line 246. This produces an OpenAI-invalid tool message. Either skip rows with empty tool_results entirely, or treat an empty list as a validation error in _validate_conversation_structure.

Fixed. Empty tool results return []

🔵 LOW · testing · [Claude]

The "tool use" integration tests verify that tools, tool_calls, and tool_results are forwarded to the endpoint and that all client turns are issued, but they only run against an EchoServer that does not validate tool_call_id pairing or generate fresh tool_call ids. As a result:

• The live-history tool_call_id mismatch issue (see related issue) cannot be detected by these tests.
• There is no test covering pipeline error propagation (today, a crash inside _conv_pipeline is silently swallowed by gather(return_exceptions=True)).
• There is no concurrent-conversation stress test (>10 conversations under a non-trivial target_concurrency) that exercises the semaphore + timeout interaction.

Add at least: (1) a test using a fake server that asserts each tool message's tool_call_id matches a prior assistant tool_calls id under live-history mode (currently expected to fail), (2) a test asserting that an exception in a pipeline propagates out of MultiTurnStrategy.execute, and (3) a stress test with 50+ concurrent conversations and target_concurrency set, asserting all complete and the semaphore reaches its rest value.

Added a test with concurrent conversations and multiple turns.

🔵 LOW · performance · [Claude]

_validate_conversation_structure, _validate_turn_numbering, and _build_metadata each call self.dataframe.groupby("conversation_id") independently (lines 142, 164, 201, 280). For a dataset with thousands of conversations, these are repeated O(N log N) operations on top of a to_dict(orient="records") walk in _validate_conversation_grouping. Cache the groupby once at the top of __init__ and pass the grouped object into each helper. This is dataset-load-time only (cold path), so it's a low-priority cleanup, but the redundancy is currently visible in the code.

Fixed

🔵 LOW · concurrency · [Claude]

turn_done: asyncio.Event = field(default_factory=asyncio.Event) is constructed at ConversationState instantiation time, before MultiTurnStrategy.execute runs. In Python 3.10+ this no longer requires a running event loop at construction time, but the resulting Event is still bound to whichever loop first awaits/sets it. MultiTurnStrategy.__init__ similarly constructs asyncio.Semaphore(target_concurrency) (multi_turn_strategy.py line 84). Today _run_benchmark_async happens to construct both inside the eventual benchmark loop, but that's fragile — anyone who instantiates a MultiTurnStrategy outside the loop (tests, CLI helpers, future REPL) will silently get cross-loop primitives that hang on wait(). Defer creation to execute() (where the running loop is guaranteed to exist) or assert asyncio.get_running_loop() in __init__.

🔵 LOW · design · [Claude]

self._inflight: dict[str, str] = {} is mutated from two execution contexts (the per-conversation pipeline tasks at line 190, and on_sample_complete invoked from the recv task at line 218) and is never bounded. While Python's GIL makes the dict ops atomic and single-threaded asyncio prevents true concurrent access, every entry that doesn't get a response (timeouts, dropped responses on shutdown, etc.) leaks for the lifetime of the strategy. For a long-running benchmark with many timeouts, this is a silent memory/diagnosability issue. Consider clearing _inflight entries explicitly in the timeout branch (line 165) and on session-stop, and add a debug-time assertion that _inflight is empty when execute returns.

+1

_build_metadata() currently appends sample entries with "index": idx where idx is the DataFrame row index from iterrows(). That index is not guaranteed to be a dense 0..N-1 mapping to Dataset.load_sample() indices (especially after filtering to client turns). To avoid incorrect sample issuance, store a dense sample_index that matches the position in self.data (client_turn_samples) and have the scheduler use that for PhaseIssuer.issue().

execute() uses enumerate(dataset_metadata["samples"]) to derive sample_index for phase_issuer.issue(). This assumes the metadata list order exactly matches the dataset's load_sample() indexing. With MultiTurnDataset, that coupling is fragile (and can break if metadata ordering differs from self.data). Prefer reading an explicit sample_index field from metadata (or another stable mapping) that is guaranteed to refer to the dataset sample index.

_expand_tool_results() will emit tool messages even if a tool_results entry is missing tool_call_id or content (they become None). That produces invalid OpenAI wire-format messages and will likely fail downstream. Since tool rows are required to have non-empty tool_results with required fields (per schema/docs), consider validating each entry and raising InputValidationError when required keys are absent.

MultiTurnDataset builds self._conv_groups via dataframe.groupby("conversation_id"), which defaults to sorting keys. That can reorder conversations compared to file order, while load() later builds self.data in raw row order. If conversation_ids are not already lexicographically sorted, conversation_metadata["samples"] ordering can diverge from load_sample() indices, causing MultiTurnStrategy to issue the wrong samples/turns. Consider using groupby(..., sort=False) and/or explicitly constructing a dense sample_index in file order that is guaranteed to match self.data indices.

What are the metrics for these suggestions? For Small (< 50 convs) does a concurrency of 32 give better performance? I would assume the model size/server capability also plays a role in the concurrency settings.

Removed this part. Only keeping concurrency definition.

clarify with "conversations with large number of turns" - a long conversation can also mean small number of long turns as end to end time could also be large there.

Same as above. Removed this exact phrasing.

Is it possible to add a utility that parses the dataset to make sure it is compliant so devs can use it instead of running the benchmark for testing.

Added documentation to use scripts/validate_jsonl_schema.py for validation

+1

So one task for each conversation?

Is this related to the multi-turn testing?

Yes, more specifically for tool call handling

This is something that needs fixing/clarification. As-is, this would mean that we have multiple conversations open, with N of them (N being the concurrency) sending requests. If i understand correctly, we only want to have N conversations open, with a turn from each currently being processed.
Launching tasks for each conversation will have this drawback - we cannot schedule tasks from a subset of conversations as all conversations will be available to run.

Refactored to advance turn on on_sample_complete. Only N conversations are kept open, and a new conversation only starts when an existing one finishes.